﻿/*
   Copyright 2013 Dmitry Bratus

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Jurassic.Library;
using Jurassic;
using Flower.Directory.Scripting;
using Flower.Services;
using Flower.Services.Data;
using System.Transactions;
using System.Threading;
using Flower.Directory.Util;

namespace Flower.Directory.Management.Scripting
{
    internal class DecommissionProcessor : ClrFunction
    {
        private const int ProcessesPerBatch = 100;
        private const int RecipientsPerBatch = 10;
        private static readonly string[] FoldersWithProcessesToMove =
            new[] { "Pending", "Running", "Waiting" };

        public DecommissionProcessor(ScriptEngine engine)
            : base
            (
                engine.Function.InstancePrototype,
                "decommissionProcessor",
                engine.Object.InstancePrototype
            )
        {
        }

        [JSCallFunction]
        public void Exec(string processorName, object options)
        {
            try
            {
                bool silent = false;
                int transactionTimeoutSec = 30;
                int queuesMigrationDelaySec = 1;
                int reportStatusEverySec = 1;
                bool startMigrated = true;

                var optionsObj = options as ObjectInstance;

                if (optionsObj != null)
                {
                    GetBool(optionsObj, "silent", ref silent);
                    GetInt(optionsObj, "transactionTimeoutSec", ref transactionTimeoutSec);
                    GetInt(optionsObj, "queuesMigrationDelaySec", ref queuesMigrationDelaySec);
                    GetInt(optionsObj, "reportStatusEverySec", ref reportStatusEverySec);
                    GetBool(optionsObj, "startMigrated", ref startMigrated);
                }

                var trnOptions = new TransactionOptions
                {
                    IsolationLevel = IsolationLevel.ReadCommitted,
                    Timeout = TimeSpan.FromSeconds(transactionTimeoutSec)
                };

                var dirObj = Engine.GetGlobalValue<ObjectInstance>("$dir") as DirectoryObject;

                if (dirObj == null)
                {
                    throw new JavaScriptException(Engine, "Error", "No directory object found.");
                }

                IDirectory dir = dirObj.Directory;

                if (!silent)
                {
                    Console.WriteLine("Deactivating processor.");
                }

                dir.Remove("/Processors/Active/" + processorName);

                if (!silent)
                {
                    Console.WriteLine("Getting processes to migrate.");
                }

                List<Process> processes =
                    GetProcessorEntries(dir, processorName, FoldersWithProcessesToMove)
                        .OfType<Process>()
                        .ToList(); // All processes must be loaded before being moved.

                if (!silent)
                {
                    Console.WriteLine("Got {0} processes to migrate.", processes.Count);
                }

                var pidToProcessorMap =
                    new Dictionary<string, string>();

                DateTime statusReportedAt = DateTime.Now, now;
                int cnt = 0;

                foreach (Process process in processes)
                {
                    using (var trn = new TransactionScope(TransactionScopeOption.Required, trnOptions))
                    {
                        string targetProcessor = SelectProcessor(dir, process.Workflow);
                        
                        if (targetProcessor == null)
                        {
                            throw new JavaScriptException(Engine, "Error", string.Format("No active processors to distribute process {0} to.", process.Id));
                        }
                        
                        dir.Move(process.Id, string.Concat("/Processors/", targetProcessor, "/Migrated"));
                        
                        pidToProcessorMap.Add(process.Id, targetProcessor);

                        trn.Complete();
                    }

                    cnt++;

                    if (!silent)
                    {
                        now = DateTime.Now;
                        if (now - statusReportedAt >= TimeSpan.FromSeconds(reportStatusEverySec))
                        {
                            Console.WriteLine("{0} processes migrated.", cnt);
                            
                            statusReportedAt = now;
                        }
                    }
                }

                if (!silent)
                {
                    Console.WriteLine("{0} processes migrated.", cnt);
                }

                using (var trn = new TransactionScope(TransactionScopeOption.Required, trnOptions))
                {
                    if (!silent)
                    {
                        Console.WriteLine("Getting suspention links to remove.");
                    }

                    List<DirectoryEntry> suspended =
                        GetProcessorEntries(dir, processorName, new[] { "Suspended" })
                            .ToList(); // All links must be loaded before being removed.
                    
                    if (!silent)
                    {
                        Console.WriteLine("Got {0} suspention links to remove.", suspended.Count);
                    }

                    statusReportedAt = DateTime.Now;
                    cnt = 0;

                    foreach (DirectoryEntry link in suspended)
                    {
                        dir.Remove(link.Id);

                        cnt++;

                        if (!silent)
                        {
                            now = DateTime.Now;
                            if (now - statusReportedAt >= TimeSpan.FromSeconds(reportStatusEverySec))
                            {
                                Console.WriteLine("{0} suspention links removed.", cnt);

                                statusReportedAt = now;
                            }
                        }
                    }

                    if (!silent)
                    {
                        Console.WriteLine("{0} suspention links removed.", cnt);
                    }

                    trn.Complete();
                }

                if (!silent)
                {
                    Console.WriteLine("Waiting for {0} before messages migration.", TimeSpan.FromSeconds(queuesMigrationDelaySec));
                }

                Thread.Sleep(TimeSpan.FromSeconds(queuesMigrationDelaySec));

                if (!silent)
                {
                    Console.WriteLine("Migrating messages.");
                }

                var processor = (Processor)dir.Get("/Processors/" + processorName);
                var queue = DirectoryUtil.GetService<IQueueService>(processor, "queue");

                long offset = 0;
                long limit = RecipientsPerBatch;
                IList<string> recipientIds;
                var processorsQueues = new Dictionary<string, IQueueService>();

                statusReportedAt = DateTime.Now;
                cnt = 0;

                while ((recipientIds = queue.GetRecipients(offset, limit)).Count > 0)
                {
                    using (var trn = new TransactionScope(TransactionScopeOption.Required, trnOptions))
                    {
                        var recipients = queue.ExportRecipients(recipientIds);

                        var recipientsByProcessors =
                            recipients
                                .Where(r => pidToProcessorMap.ContainsKey(PidFromRecipientId(r.Id)))
                                .GroupBy(r => pidToProcessorMap[PidFromRecipientId(r.Id)], r => r);

                        foreach (var procRcp in recipientsByProcessors)
                        {
                            IQueueService targetQueue;

                            if (!processorsQueues.TryGetValue(procRcp.Key, out targetQueue))
                            {
                                var targetProc = (Processor)dir.Get("/Processors/" + procRcp.Key);
                                targetQueue = DirectoryUtil.GetService<IQueueService>(targetProc, "queue");
                                processorsQueues.Add(procRcp.Key, targetQueue);
                            }

                            targetQueue.ImportRecipients(procRcp.ToList());
                        }

                        offset = limit;
                        limit += RecipientsPerBatch;

                        trn.Complete();
                    }

                    cnt += recipientIds.Count;

                    if (!silent)
                    {
                        now = DateTime.Now;
                        if (now - statusReportedAt >= TimeSpan.FromSeconds(reportStatusEverySec))
                        {
                            Console.WriteLine("{0} recipients migrated.", cnt);

                            statusReportedAt = now;
                        }
                    }
                }

                if (!silent)
                {
                    Console.WriteLine("{0} recipients migrated.", cnt);
                }

                if (startMigrated)
                {
                    if (!silent)
                    {
                        Console.WriteLine("Resuming {0} migrated processes.", pidToProcessorMap.Count);
                    }

                    statusReportedAt = DateTime.Now;
                    cnt = 0;

                    foreach (var pidProc in pidToProcessorMap)
                    {
                        dir.Move(pidProc.Key, string.Concat("/Processors/", pidProc.Value, "/Pending"));

                        cnt++;

                        if (!silent)
                        {
                            now = DateTime.Now;
                            if (now - statusReportedAt >= TimeSpan.FromSeconds(reportStatusEverySec))
                            {
                                Console.WriteLine("{0} processes resumed.", cnt);

                                statusReportedAt = now;
                            }
                        }
                    }

                    if (!silent)
                    {
                        Console.WriteLine("{0} processes resumed.", cnt);
                    }
                }
            }
            catch (JavaScriptException)
            {
                throw;
            }
            catch (Exception ex)
            {
                throw new JavaScriptException(Engine, "Error", ex.Message, ex);
            }
        }

        private static string PidFromRecipientId(string recipientId)
        {
            int slashIdx = recipientId.IndexOf('/');
            return (slashIdx > 0) ? recipientId.Substring(0, slashIdx) : string.Empty;
        }

        private static bool GetInt(ObjectInstance obj, string attr, ref int val)
        {
            object valObj = obj[attr];

            if (valObj == null)
            {
                return false;
            }

            return int.TryParse(valObj.ToString(), out val);
        }

        private static void GetBool(ObjectInstance obj, string attr, ref bool val)
        {
            object valObj = obj[attr];

            if (valObj == null)
            {
                return;
            }

            if (valObj is bool)
            {
                val = (bool)valObj;
            }

            var valBool = valObj as BooleanInstance;

            if (valBool != null)
            {
                val = valBool.Value;
            }

            int intVal = 0;

            if (GetInt(obj, attr, ref intVal))
            {
                val = intVal != 0;
            }
        }

        private static IEnumerable<DirectoryEntry> GetProcessorEntries(IDirectory dir, string processorName, IEnumerable<string> folders)
        {
            foreach (string fld in folders)
            {
                string parentPath = string.Concat("/Processors/", processorName, "/", fld);

                long offset = 0;
                long limit = ProcessesPerBatch;
                IList<DirectoryEntry> processes;

                while 
                (
                    (
                        processes = 
                            dir.GetChildren
                            (
                                parentPath, 
                                DirectoryEntryTypes.Process, 
                                SortProperty.Default, 
                                SortOrder.Asc, 
                                offset, 
                                limit
                            )
                    ).Count > 0
                )
                {
                    foreach (DirectoryEntry prc in processes)
                    {
                        yield return prc;
                    }

                    offset = limit;
                    limit += ProcessesPerBatch;
                }
            }
        }

        private static string SelectProcessor(IDirectory dir, string workflow)
        {
            ICollection<ScriptParameter> outParams = dir.ExecuteScript
            (
                "@SelectProcessor",
                new List<ScriptParameter>
                {
                    new StringScriptParameter
                    {
                        Name = "workflow",
                        Value = workflow
                    }
                }
            );

            ScriptParameter processorNameParam = 
                outParams.FirstOrDefault(p => p.Name == "processor");

            return (processorNameParam != null) ? ((StringScriptParameter)processorNameParam).Value : null;
        }
    }
}
